package org.ws.task;

import com.google.common.collect.Maps;
import io.netty.util.concurrent.DefaultThreadFactory;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.ws.task.common.AddressUtils;
import org.ws.task.exception.NoEventActionException;
import org.ws.task.zookeeper.DefaultZkNodeManager;
import org.ws.task.zookeeper.ZkDataManager;
import org.ws.task.zookeeper.ZkNodeManager;
import org.ws.task.event.EventListener;
import org.ws.task.event.support.startup.BootstrapEvent;
import org.ws.task.event.support.startup.BootstrapEventType;
import org.ws.task.event.support.BootstrapEventHolder;
import org.ws.task.model.NodeStatus;
import org.ws.task.repository.LeaderContextRepository;
import org.ws.task.repository.NodeMetadataRepository;
import org.ws.task.repository.TaskRepository;
import org.ws.task.model.NodeMetadata;
import org.ws.task.model.TaskConfig;
import org.ws.task.netty.NettyClient;
import org.ws.task.netty.NettyConfig;
import org.ws.task.netty.NettyServer;
import org.ws.task.netty.support.DefaultNettyClient;
import org.ws.task.netty.support.DefaultNettyServer;
import org.ws.task.zookeeper.ZkConfig;
import org.ws.task.listener.zk.CompleteNodeListener;

import java.util.Collection;
import java.util.Map;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;

import static org.ws.task.event.support.startup.BootstrapEventType.BECOME_LEADER;
import static org.ws.task.event.support.startup.BootstrapEventType.CHANGE_LEADER;
import static org.ws.task.event.support.startup.BootstrapEventType.CHANNEL_CLOSE;
import static org.ws.task.event.support.startup.BootstrapEventType.CHANNEL_CONNECT;
import static org.ws.task.event.support.startup.BootstrapEventType.CHANNEL_IDLE;
import static org.ws.task.event.support.startup.BootstrapEventType.CHECK_LEADER;
import static org.ws.task.event.support.startup.BootstrapEventType.CLIENT_CHANNEL_CONNECT;
import static org.ws.task.event.support.startup.BootstrapEventType.CLIENT_CHANNEL_IDLE;
import static org.ws.task.event.support.startup.BootstrapEventType.CLIENT_STARTUP;
import static org.ws.task.event.support.startup.BootstrapEventType.NOT_FOUND_LEADER;
import static org.ws.task.event.support.startup.BootstrapEventType.NOT_LEADER;
import static org.ws.task.event.support.startup.BootstrapEventType.REMOVE_LEADER;
import static org.ws.task.event.support.startup.BootstrapEventType.REMOVE_TASK_CONFIG;
import static org.ws.task.event.support.startup.BootstrapEventType.REMOVE_WORKER;
import static org.ws.task.event.support.startup.BootstrapEventType.SERVER_STARTUP;
import static org.ws.task.event.support.startup.BootstrapEventType.UPDATE_TASK_CONFIG;
import static org.ws.task.event.support.startup.BootstrapEventType.UPDATE_WORKER;

/**
 * 启动入口
 */
public final class TaskMainBootstrap implements Bootable,AutoCloseable{

    private static Logger log = LoggerFactory.getLogger(TaskMainBootstrap.class.getName());

    /**
     * 命名空间，每一个APP对应唯一一个，在ZK中所有元数据在{namespace}下
     */
    private final String namespace;
    /**
     * ZK节点管理器，元数据同步管理
     */
    private final ZkNodeManager zkNodeManager;
    /**
     * 节点通讯Server端：当前节点为Leader时，需启动Server端
     */
    private final NettyServer nettyServer;
    /**
     * 节点通讯Client端：任一节点都启动Client端，长连接保持与Server端的通讯
     */
    private final NettyClient nettyClient;
    /**
     * Leader端控制器
     */
    private final LeaderController leaderController;
    /**
     * Worker端控制器
     */
    private final WorkerController workerController;
    /**
     * 心跳定时器：NettyClient向NettyServer发送心跳
     */
    private final Timer clientHeartbeatTimer;
    /**
     * 心跳定时器是否已经在运行标记
     */
    private final AtomicBoolean clientHeartbeatTimerRunning = new AtomicBoolean(false);
    /**
     * 节点启动事件处理器（启动完成后节点出现异常，仍由事件触发后续操作调整到健康状态）
     */
    private final BootstrapEventHandler bootstrapEventHandler;
    /**
     * 节点启动事件Listener，包括Leader，Worker各个事件触发的操作
     */
    private final BootstrapNodeListener bootstrapNodeListener;
    /**
     * 启动事件执行器
     */
    private final BootstrapEventExecutor bootstrapEventExecutor;
    /**
     * 节点启动异步操作线程池
     */
    private final ExecutorService executor;
    /**
     * 第一次启动成功标识
     */
    private final AtomicBoolean firstSuccess = new AtomicBoolean(false);

    /**
     * 根据ZkConfig和NettyConfig创建启动器
     * @param zkConfig
     * @param nettyConfig
     */
    public TaskMainBootstrap(final ZkConfig zkConfig, final NettyConfig nettyConfig){
        // 验证配置
        configValidated(zkConfig,nettyConfig);

        this.namespace = zkConfig.getNamespace();
        this.zkNodeManager = new DefaultZkNodeManager(zkConfig);
        this.nettyServer = new DefaultNettyServer(nettyConfig);
        this.nettyClient = new DefaultNettyClient(nettyConfig);
        this.leaderController = new LeaderController(this.nettyServer,nettyConfig);
        this.workerController = new WorkerController(this.nettyClient,nettyConfig);
        this.executor = Executors.newFixedThreadPool(8,new DefaultThreadFactory("MainBootstrap"));
        this.clientHeartbeatTimer = new Timer("ClientKeepingTimer", false);
        this.bootstrapEventHandler = new BootstrapEventHandler();
        this.bootstrapNodeListener = new BootstrapNodeListener();
        this.bootstrapEventExecutor = new BootstrapEventExecutor();
        // 注册启动事件触发操作
        this.registerBootstrapEventAction();
    }

    /**
     * 验证配置
     * @param zkConfig
     * @param nettyConfig
     */
    private void configValidated(final ZkConfig zkConfig, final NettyConfig nettyConfig){
        if(zkConfig == null){
            throw new IllegalArgumentException("zkConfig must be not null.");
        }
        if(nettyConfig == null){
            throw new IllegalArgumentException("nettyConfig must be not null.");
        }
        if(StringUtils.isBlank(zkConfig.getNamespace())){
            throw new IllegalArgumentException("namespace must be not blank.");
        }
        if(StringUtils.isBlank(zkConfig.getAddress())){
            throw new IllegalArgumentException("zookeeper address must be not blank.");
        }
        if(nettyConfig.getServerPort() <= 0){
            throw new IllegalArgumentException("netty server port must be gt 0.");
        }
    }

    @Override
    public void start() {
        // 注册事件处理器
        BootstrapEventHolder.getInstance().registerListener(this.bootstrapEventHandler);
        // 启动节点管理器：竞选Leader，同步元数据
        TaskMainBootstrap.this.zkNodeManager.start();
        // 启动Leader控制器：Leader节点才起作用
        TaskMainBootstrap.this.leaderController.start();
        // 启动Worker管理器
        TaskMainBootstrap.this.workerController.start();
        log.info("MainBootstrap starting with namespace[{}].",this.namespace);
    }

    @Override
    public void shutdown() {
        this.zkNodeManager.shutdown();
        this.workerController.shutdown();
        this.leaderController.shutdown();
        this.nettyClient.shutdown();
        this.nettyServer.shutdown();
    }

    @Override
    public void close() throws Exception {
        shutdown();
    }

    /**
     * 获取Zk元数据
     * @return
     */
    public ZkDataManager getZkDataManager(){
        return this.zkNodeManager;
    }

    /**
     * 验证Node状态。
     * 1.检查元数据Leader状态和port验证Leader是否有效；
     * 2.Leader节点通过检查NettyServer的状态和port验证Server是否有效；
     * 3.检查元数据所有的Worker状态和port验证所有Worker是否有效；
     * 4.检查当前Worker状态和port验证当前Worker是否有效；
     * 5.检查NettyClient的状态和port验证NettyClient是否有效；
     *
     * 当前节点状态调整：
     * 1.若NettyClient已启动，worker状态未更新-》更新本节点状态；
     * 2.leader已启动，客户端未启动成功，清理channel重新链接；
     *
     * 在第一次所有的验证全部通过则发生启动成功事件。
     */
    private void checkNodeStatus(){
        // 节点整体状态
        boolean successful = true;
        // leader有效状态
        boolean leaderActive = true;
        // worker有效状态
        boolean workerActive = true;
        // 检查Leader
        NodeMetadata leader = NodeMetadataRepository.getInstance().getLeader();
        if (!leader.isActive()) {
            log.warn("leader is not active[{}].",leader.getStatus());
            leaderActive = false;
            successful = false;
        }
        else {
            log.debug("leader is active[{}].",leader.getStatus());
        }
        if(leader.getPort()<0){
            log.warn("leader port[{}] error.",leader.getPort());
            leaderActive = false;
            successful = false;
        }
        else {
            log.debug("leader port[{}] ok.",leader.getPort());
        }
        // 本节点为Leader，检查netty server
        if(this.zkNodeManager.isLeader()){
            if (!this.nettyServer.isActive()) {
                log.warn("netty server is not active.");
                successful = false;
            }
            else {
                log.debug("netty server is active.");
            }
            if(this.nettyServer.port()<0){
                log.warn("netty server port[{}] error.",this.nettyServer.port());
                successful = false;
            }
            else {
                log.debug("netty server port[{}] ok.",this.nettyServer.port());
            }
            // leader节点状态不正确，netty server启动成功-》更新节点状态
            if(!leaderActive && this.nettyServer.isActive() && this.nettyServer.port()>0){
                this.zkNodeManager.updateLeader(this.nettyServer.port(),NodeStatus.ACTIVE);
            }
        }

        // worker
        Collection<NodeMetadata> workers = NodeMetadataRepository.getInstance().getWorkers();
        if(CollectionUtils.isNotEmpty(workers)){
            for (NodeMetadata worker : workers) {
                if(!worker.isActive()){
                    log.warn("worker[{}] is not active[{}].",worker.getId(),worker.getStatus());
                    successful = false;
                }
                else {
                    log.debug("worker[{}] is active[{}].",worker.getId(),worker.getStatus());
                }
                if(worker.getPort()<0){
                    log.warn("worker[{}] port[{}] error.",worker.getId(),worker.getPort());
                    successful = false;
                }
                else {
                    log.debug("worker[{}] port[{}] ok.",worker.getId(),worker.getPort());
                }
            }
        }
        NodeMetadata worker = NodeMetadataRepository.getInstance().getTheWorkerNode();
        if(worker != null){
            if(!worker.isActive()){
                log.warn("this worker[{}] is not active[{}].",worker.getId(),worker.getStatus());
                workerActive = false;
                successful = false;
            }
            else {
                log.debug("this worker[{}] is active[{}].",worker.getId(),worker.getStatus());
            }
            if(worker.getPort()<0){
                log.warn("this worker[{}] port[{}] error.",worker.getId(),worker.getPort());
                workerActive = false;
                successful = false;
            }
            else {
                log.debug("this worker[{}] port[{}] ok.",worker.getId(),worker.getPort());
            }
        }
        if(!this.nettyClient.isActive()){
            log.warn("netty client is not active.");
            successful = false;
        }
        else {
            log.debug("netty client is active.");
        }
        if(this.nettyClient.port()<0){
            log.warn("netty client port[{}] error.",this.nettyClient.port());
            successful = false;
        }
        else {
            log.debug("netty client port[{}] ok.",this.nettyClient.port());
        }
        // 若netty client已启动，worker状态未更新-》更新本节点状态
        if(!workerActive && this.nettyClient.isActive() && this.nettyClient.port()>0){
            this.zkNodeManager.updateWorker(this.nettyClient.port(),NodeStatus.ACTIVE);
        }
        // leader已启动，客户端未启动成功，清理channel重新链接
        else if(leaderActive
                && this.nettyClient.isActive()
                && this.nettyClient.port()<0){
            try {
                this.nettyClient.clearChannel();
            } catch (Exception e) {
                log.error(e.getMessage(),e);
            }
            this.nettyClient.heartbeat();
        }
        if(successful){
            log.info("check end: leader&worker node and netty all successfully.");
            if(!firstSuccess.get()){
                // 只在第一次成功时发送
                BootstrapEventHolder.getInstance().publishEvent(new BootstrapEvent(true, BootstrapEventType.BOOTSTRAP_SUCCESS));
                firstSuccess.set(true);
            }
        }
        else {
            log.warn("check end: leader&worker node or netty failed.");
        }
    }

    /**
     * 保持客户端心跳
     */
    private void clientHeartbeat(){
        // 已经运行，不重复执行
        if(this.clientHeartbeatTimerRunning.get()){
            return;
        }
        this.clientHeartbeatTimerRunning.set(true);
        // 每10秒发送客户端心跳
        this.clientHeartbeatTimer.scheduleAtFixedRate(new TimerTask() {
            @Override
            public void run() {
                try {
                    TaskMainBootstrap.this.nettyClient.heartbeat();
                } catch (Exception e) {
                    log.error(e.getMessage(),e);
                }
            }
        }, 10_000, 10_000);
    }

    /**
     * 注册启动事件触发的操作处理策略
     */
    private void registerBootstrapEventAction(){
        this.bootstrapEventExecutor
                // 成为Leader
                .registerAction(BECOME_LEADER,o -> this.bootstrapNodeListener.becomeLeader((NodeMetadata) o))
                // 不是Leader
                .registerAction(NOT_LEADER,o->this.bootstrapNodeListener.notLeader((NodeMetadata) o))
                // 变更Leader
                .registerAction(CHANGE_LEADER,o->this.bootstrapNodeListener.changeLeaderNode((NodeMetadata) o))
                // 删除Leader
                .registerAction(REMOVE_LEADER,o->this.bootstrapNodeListener.removeLeaderNode((NodeMetadata) o))
                // 检查Leader状态
                .registerAction(CHECK_LEADER,o-> TaskMainBootstrap.this.checkNodeStatus())
                // 更新Worker
                .registerAction(UPDATE_WORKER,o->this.bootstrapNodeListener.updateWorkerNode((NodeMetadata) o))
                // 删除Worker
                .registerAction(REMOVE_WORKER,o->this.bootstrapNodeListener.removeWorkerNode((NodeMetadata) o))
                // 服务端已启动
                // 当前节点为Leader：等待服务端启动后启动客户端
                .registerAction(SERVER_STARTUP,o->{
                    int port = (int)o;
                    TaskMainBootstrap.this.zkNodeManager.updateLeader(port,NodeStatus.ACTIVE);
                    NodeMetadata nodeMetadata = TaskMainBootstrap.this.zkNodeManager.getLeader();
                    NodeMetadataRepository.getInstance().changeLeader(nodeMetadata);
                    // 客户端设置Leader地址
                    TaskMainBootstrap.this.nettyClient.setLeaderAddress(nodeMetadata.getAddress());
                    TaskMainBootstrap.this.nettyClient.start();
                })
                // 服务端：已连接
                .registerAction(CHANNEL_CONNECT,o->{
                    final String addr = (String)o;
                    final int port = AddressUtils.parsePort(addr);
                    NodeMetadataRepository.getInstance().getWorkers().stream().forEach(nodeMetadata -> {
                        if(nodeMetadata != null && StringUtils.equals(addr,nodeMetadata.getAddress())){
                            log.debug("update node -> {}",nodeMetadata.toString());
                            nodeMetadata.setStatus(NodeStatus.ACTIVE);
                            TaskMainBootstrap.this.zkNodeManager.updateWorker(port,NodeStatus.ACTIVE);
                        }
                    });
                })
                // 服务端：已关闭
                .registerAction(CHANNEL_CLOSE,o->{
                    // TODO：根据addr无法匹配到旧的节点
                    final String addr = (String)o;
                    NodeMetadataRepository.getInstance().getWorkers().stream().forEach(nodeMetadata -> {
                        if(nodeMetadata != null && StringUtils.equals(addr,nodeMetadata.getAddress())){
                            log.debug("update node -> {}",nodeMetadata.toString());
                            nodeMetadata.setStatus(NodeStatus.GONE);
                            TaskMainBootstrap.this.zkNodeManager.updateWorker(-1,NodeStatus.GONE);
                        }
                    });
                })
                // 服务端：已超时
                .registerAction(CHANNEL_IDLE,o->{
                    final String addr = (String)o;
                    final int port = AddressUtils.parsePort(addr);
                    NodeMetadataRepository.getInstance().getWorkers().stream().forEach(nodeMetadata -> {
                        if(nodeMetadata != null && StringUtils.equals(addr,nodeMetadata.getAddress())){
                            log.debug("update node -> {}",nodeMetadata.toString());
                            nodeMetadata.setStatus( NodeStatus.RECOVERING);
                            TaskMainBootstrap.this.zkNodeManager.updateWorker(port,NodeStatus.RECOVERING);
                        }
                    });
                })
                // 更新任务配置
                .registerAction(UPDATE_TASK_CONFIG,o->this.bootstrapNodeListener.updateTaskConfig((TaskConfig) o))
                // 删除任务配置
                .registerAction(REMOVE_TASK_CONFIG,o->this.bootstrapNodeListener.removeTaskConfig((TaskConfig) o))
                // 客户端已启动
                .registerAction(CLIENT_STARTUP,o->{
                    // 客户端启动成功后：启动worker节点
                    int port = (int)o;
                    TaskMainBootstrap.this.zkNodeManager.updateWorker(port,NodeStatus.ACTIVE);
                    // 启动客户端心跳
                    TaskMainBootstrap.this.clientHeartbeat();
                })
                // 客户端：已连接
                .registerAction(CLIENT_CHANNEL_CONNECT,o->{
                    // 客户端建立新链接后，若port变更则更新worker节点
                    int port = (int)o;
                    // 启动客户端心跳
                    TaskMainBootstrap.this.clientHeartbeat();
                    TaskMainBootstrap.this.zkNodeManager.updateWorker(port,NodeStatus.ACTIVE);
                    TaskMainBootstrap.this.checkNodeStatus();
                })
                // 客户端：已超时
                .registerAction(CLIENT_CHANNEL_IDLE,o->{
                    // 当前客户端超时：再次尝试启动心跳
                    int port = TaskMainBootstrap.this.nettyClient.port();
                    TaskMainBootstrap.this.zkNodeManager.updateWorker(port,NodeStatus.RECOVERING);
                    TaskMainBootstrap.this.clientHeartbeat();
                })
                // 客户端：未获取到Leader
                .registerAction(NOT_FOUND_LEADER,o-> TaskMainBootstrap.this.checkNodeStatus());
    }

    /**
     * 启动事件处理器
     */
    class BootstrapEventHandler implements EventListener<BootstrapEvent>, CompleteNodeListener {
        // 使用子线程执行事件触发后续操作
        private final ExecutorService executorService = Executors.newSingleThreadExecutor(new DefaultThreadFactory("BootstrapEventExecutor"));

        @Override
        public void onEvent(BootstrapEvent event) {
            executorService.submit(()->{
                BootstrapEventType eventType = event.getEventType();
                Object source = event.getSource();
                TaskMainBootstrap.this.bootstrapEventExecutor.accept(eventType,source);
            });
        }

    }

    /**
     * NodeListener启动处理
     *
     */
    class BootstrapNodeListener implements CompleteNodeListener{
        @Override
        public void becomeLeader(NodeMetadata nodeMetadata) {
            TaskMainBootstrap.this.executor.submit(()->{
                // 启动服务端
                TaskMainBootstrap.this.nettyServer.start();
                // 更新Leader节点数据
                NodeMetadataRepository.getInstance().changeLeader(nodeMetadata);
            });
        }

        @Override
        public void notLeader(NodeMetadata nodeMetadata) {
            TaskMainBootstrap.this.executor.submit(()->{
                // 客户端设置Leader地址
                TaskMainBootstrap.this.nettyClient.setLeaderAddress(nodeMetadata.getAddress());
                // 更新Leader节点数据
                NodeMetadataRepository.getInstance().changeLeader(nodeMetadata);
                // 只启动客户端
                TaskMainBootstrap.this.nettyClient.start();
            });
        }

        @Override
        public void changeLeaderNode(NodeMetadata nodeMetadata) {
            // 客户端设置Leader地址
            TaskMainBootstrap.this.nettyClient.setLeaderAddress(nodeMetadata.getAddress());
            // 更新Leader节点数据
            NodeMetadataRepository.getInstance().changeLeader(nodeMetadata);
            TaskMainBootstrap.this.checkNodeStatus();
        }

        @Override
        public void removeLeaderNode(NodeMetadata nodeMetadata) {
            // 移除Leader信息
            TaskMainBootstrap.this.nettyClient.setLeaderAddress(null);
            nodeMetadata.setStatus(NodeStatus.GONE);
            NodeMetadataRepository.getInstance().changeLeader(nodeMetadata);
        }

        @Override
        public void updateTaskConfig(TaskConfig taskConfig) {
            // TODO：
        }

        @Override
        public void removeTaskConfig(TaskConfig taskConfig) {
            // TODO：
        }

        @Override
        public void updateWorkerNode(NodeMetadata nodeMetadata) {
            // 更新worker：新增或更新
            log.debug("update worker node -> {}",nodeMetadata.toString());
            NodeMetadata node = NodeMetadataRepository.getInstance().getNode(nodeMetadata.getId());
            // 无变化不更新
            if(node != null && node.equals(nodeMetadata)){
                log.debug("nodeMetadata no change.");
                return;
            }
            NodeMetadataRepository.getInstance().updateWorker(nodeMetadata);
            TaskMainBootstrap.this.checkNodeStatus();
        }

        @Override
        public void removeWorkerNode(NodeMetadata nodeMetadata) {
            // 移除worker
            NodeMetadataRepository.getInstance().removeWorker(nodeMetadata);
            // 删除一个节点对task做全部的resize
            if(TaskMainBootstrap.this.zkNodeManager.isLeader()) {
                Set<String> taskNames = TaskRepository.getInstance().taskNames();
                for (String taskName : taskNames) {
                    LeaderContextRepository.getInstance().resize(taskName);
                }
            }
            TaskMainBootstrap.this.checkNodeStatus();
        }
    }

    /**
     * 启动事件处理器
     */
    static class BootstrapEventExecutor {
        // 事件类型对应操作函数
        private final Map<BootstrapEventType, Consumer> eventActionMap = Maps.newConcurrentMap();

        /**
         * 注册事件对应的操作
         * @param eventType
         * @param action
         * @return
         */
        public BootstrapEventExecutor registerAction(final BootstrapEventType eventType, Consumer<Object> action){
            if(eventActionMap.containsKey(eventType)){
                throw new NoEventActionException(eventType+" has mapped action!");
            }
            eventActionMap.put(eventType,action);
            return this;
        }

        /**
         * 执行操作
         * @param eventType
         * @param arg
         * @return
         */
        public Consumer accept(final BootstrapEventType eventType,Object arg){
            Consumer action = eventActionMap.get(eventType);
            if(action == null){
                throw new NoEventActionException(eventType+" can not mapping action.");
            }
            action.accept(arg);
            return action;
        }


    }
}
